/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
 * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
 * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
 * License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 */
package org.apache.kafka.clients;

import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.internals.ClusterResourceListeners;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.errors.TimeoutException;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

/**
 * 元数据信息
 */
public final class Metadata {


    /**
     *
     */
    public static final long TOPIC_EXPIRY_MS = 5 * 60 * 1000;
    private static final long TOPIC_EXPIRY_NEEDS_UPDATE = -1L;
    /**
     * 两个更新元数据的请求的最小的时间间隔，默认值是100ms,目的就是减少网络的压力
     */
    private final long refreshBackoffMs;
    /**
     * 多久自动更新一次元数据，默认值是5分钟更新一次。
     */
    private final long metadataExpireMs;
    /**
     * 对于producer端来讲，元数据是有版本号,每次更新元数据，都会修改一下这个版本号。
     */
    private int version;
    /**
     * 上一次更新元数据的时间。
     */
    private long lastRefreshMs;
    /**
     * 上一次成功更新元数据的时间。如果正常情况下，如果每次都是更新成功的，
     * 那么lastRefreshMs和lastSuccessfulRefreshMs 应该是相同的。
     */
    private long lastSuccessfulRefreshMs;
    /**
     * Kafka集群本身的元数据。
     */
    private Cluster cluster;
    /**
     * 这是一个标识，用来判断是否更新元数据的标识之一。
     */
    private boolean needUpdate;

    /**
     * 记录了当前已有的topics
     */
    private final Map<String, Long> topics;
    /**
     * 元数据更新监听器
     */
    private final List<Listener> listeners;
    /**
     * 集群资源监听器
     */
    private final ClusterResourceListeners clusterResourceListeners;
    private boolean needMetadataForAllTopics;
    private final boolean topicExpiryEnabled;

    /**
     * 用合理的默认设置创建一个元数据
     */
    public Metadata() {
        this(100L, 60 * 60 * 1000L);
    }

    public Metadata(long refreshBackoffMs, long metadataExpireMs) {
        this(refreshBackoffMs, metadataExpireMs, false, new ClusterResourceListeners());
    }

    /**
     * 创建一个新的元数据实例
     *
     * @param refreshBackoffMs         元数据刷新之间必须过期的最小时间,以避免繁忙,重试时间 retry.backoff.ms 默认100ms,提升服务健壮性
     *                                 polling
     * @param metadataExpireMs         元数据可以在不刷新的情况下保留的最长时间
     * @param topicExpiryEnabled       如果是真的,启用主题的不过期
     * @param clusterResourceListeners 接收元数据更新的集群ClusterResourceListeners列表
     */
    public Metadata(long refreshBackoffMs, long metadataExpireMs, boolean topicExpiryEnabled, ClusterResourceListeners clusterResourceListeners) {
        this.refreshBackoffMs = refreshBackoffMs;
        this.metadataExpireMs = metadataExpireMs;
        this.topicExpiryEnabled = topicExpiryEnabled;
        this.lastRefreshMs = 0L;
        this.lastSuccessfulRefreshMs = 0L;
        this.version = 0;
        /**
         * 这里清空了元数据
         */
        this.cluster = Cluster.empty();
        /**
         * 不需要更新,因为是第一次拉取,没有要更新的数据,只有新增
         */
        this.needUpdate = false;
        /**
         * 生产者第一次启动的时间 ,并没有主题,直接返回空的集合
         */
        this.topics = new HashMap<>();
        /**
         * 生产者第一次启动的时间 ,并没有监听,直接返回空的集合
         */
        this.listeners = new ArrayList<>();
        this.clusterResourceListeners = clusterResourceListeners;
        /**
         * 不需要
         */
        this.needMetadataForAllTopics = false;
    }

    /**
     * 从matadate中获取当前集群的信息cluster
     */
    public synchronized Cluster fetch() {
        return this.cluster;
    }

    /**
     * Add the topic to maintain in the metadata. If topic expiry is enabled, expiry time
     * will be reset on the next update.
     */
    public synchronized void add(String topic) {
        topics.put(topic, TOPIC_EXPIRY_NEEDS_UPDATE);
    }

    /**
     * The next time to update the cluster info is the maximum of the time the current info will expire and the time the
     * current info can be updated (i.e. backoff time has elapsed); If an update has been request then the expiry time
     * is now
     */
    public synchronized long timeToNextUpdate(long nowMs) {
        long timeToExpire = needUpdate ? 0 : Math.max(this.lastSuccessfulRefreshMs + this.metadataExpireMs - nowMs, 0);
        long timeToAllowUpdate = this.lastRefreshMs + this.refreshBackoffMs - nowMs;
        return Math.max(timeToExpire, timeToAllowUpdate);
    }

    /**
     * Request an update of the current cluster metadata info, return the current version before the update
     */
    public synchronized int requestUpdate() {
        this.needUpdate = true;
        return this.version;
    }

    /**
     * Check whether an update has been explicitly requested.
     *
     * @return true if an update was requested, false otherwise
     */
    public synchronized boolean updateRequested() {
        return this.needUpdate;
    }

    /**
     * 等待元数据更新,直到元数据当前的版本号大于最后一次更新的版本号
     * 可以看到,这里是线程完安全的
     *
     * @param lastVersion 上次更新的版本号
     * @param maxWaitMs   最大等待时间
     */
    public synchronized void awaitUpdate(final int lastVersion, final long maxWaitMs) throws InterruptedException {
        if (maxWaitMs < 0) {
            throw new IllegalArgumentException("Max time to wait for metadata updates should not be < 0 milli seconds");
        }
        /**
         * 获取当前时间
         */
        long begin = System.currentTimeMillis();
        /**
         * 看剩余可以使用的时间，一开始是最大等待的时间。
         */
        long remainingWaitMs = maxWaitMs;
        /**
         * version是元数据的版本号,如果当前的这个version小于等于上一次的version,说明元数据还没更新。
         * 因为如果sender线程那儿 更新元数据，如果更新成功了，sender线程肯定回去累加这个version。
         */
        while (this.version <= lastVersion) {
            /**
             * 如果还有剩余的时间。
             */
            if (remainingWaitMs != 0){
                /**
                 * 让当前线程阻塞等待。
                 * 我们这儿虽然没有去看 sender线程的源码,但是我们知道，他那儿肯定会做这样的一个操作,如果更新元数据成功了，会唤醒这个线程。
                 */
                wait(remainingWaitMs);
            }
            /**
             * 如果代码执行到这儿 说明就要么就被唤醒了，要么就到点了。
             * 计算一下花了多少时间。
             */
            long elapsed = System.currentTimeMillis() - begin;
            /**
             * 已经超时了
             */
            if (elapsed >= maxWaitMs)
            /**
             * 报一个超时的异常。
             */
                throw new TimeoutException("Failed to update metadata after " + maxWaitMs + " ms.");
            /**
             * 再次计算 可以使用的时间。
             */
            remainingWaitMs = maxWaitMs - elapsed;
        }
    }

    /**
     * Replace the current set of topics maintained to the one provided.
     * If topic expiry is enabled, expiry time of the topics will be
     * reset on the next update.
     *
     * @param topics
     */
    public synchronized void setTopics(Collection<String> topics) {
        if (!this.topics.keySet().containsAll(topics))
            requestUpdate();
        this.topics.clear();
        for (String topic : topics)
            this.topics.put(topic, TOPIC_EXPIRY_NEEDS_UPDATE);
    }

    /**
     * Get the list of topics we are currently maintaining metadata for
     */
    public synchronized Set<String> topics() {
        return new HashSet<>(this.topics.keySet());
    }

    /**
     * Check if a topic is already in the topic set.
     *
     * @param topic topic to check
     * @return true if the topic exists, false otherwise
     */
    public synchronized boolean containsTopic(String topic) {
        return this.topics.containsKey(topic);
    }

    /**
     * 更新集群元数据,如果topic过期被使用.当请求或者移出topic时,过期时间要被设置
     */
    public synchronized void update(Cluster cluster, long now) {
        /**
         * 断言cluster不能为空
         */
        Objects.requireNonNull(cluster, "cluster should not be null");
        this.needUpdate = false;
        this.lastRefreshMs = now;
        this.lastSuccessfulRefreshMs = now;
        /**
         * 我们看到了，这儿会跟新元数据的版本。
         *
         */
        this.version += 1;
        /**
         * 这个默认值是true,生产者初始化的时间给这个值赋值了,
         * Handle expiry of topics from the metadata refresh set.
         * 从刷新后的元数据中处理过期的topic数据
         */
        if (topicExpiryEnabled) {
            /**
             * 第一次进来,因为topic为空.没有下一个,不走这个遍历
             * 第二次进来就可以继续走
             */
            for (Iterator<Map.Entry<String, Long>> it = topics.entrySet().iterator(); it.hasNext(); ) {
                Map.Entry<String, Long> entry = it.next();
                /**
                 * 获取过期时间
                 */
                long expireMs = entry.getValue();
                /**
                 * 判断过期时间
                 */
                if (expireMs == TOPIC_EXPIRY_NEEDS_UPDATE)
                    entry.setValue(now + TOPIC_EXPIRY_MS);
                else if (expireMs <= now) {
                    it.remove();//移出过期
                }
            }
        }

        for (Listener listener : listeners)
            listener.onMetadataUpdate(cluster);

        String previousClusterId = cluster.clusterResource().clusterId();
        /**
         * 这个的默认值是false，所以这个分支的代码不会被运行。
         */
        if (this.needMetadataForAllTopics) {
            // the listener may change the interested topics, which could cause another metadata refresh.
            // If we have already fetched all topics, however, another fetch should be unnecessary.
            this.needUpdate = false;
            this.cluster = getClusterForCurrentTopics(cluster);
        } else {
            /**
             * 所以代码执行的是这儿。
             * 直接把刚刚传进来的对象赋值给了这个cluster。cluster代表的是kafka集群的元数据。
             * 初始化的时候，update这个方法没有去服务端拉取数据。
             */
            this.cluster = cluster;//address
        }

        // The bootstrap cluster is guaranteed not to have any useful information
        if (!cluster.isBootstrapConfigured()) {
            String clusterId = cluster.clusterResource().clusterId();
            if (clusterId == null ? previousClusterId != null : !clusterId.equals(previousClusterId))

                clusterResourceListeners.onUpdate(cluster.clusterResource());
        }
        /**
         * 大家发现这儿会有一个notifyAll，这个最重要的一个作用是不是就是唤醒，我们上一讲,看到那个wait的线程。
         */
        notifyAll();

    }

    /**
     * Record an attempt to update the metadata that failed. We need to keep track of this
     * to avoid retrying immediately.
     */
    public synchronized void failedUpdate(long now) {
        this.lastRefreshMs = now;
    }

    /**
     * @return The current metadata version
     */
    public synchronized int version() {
        return this.version;
    }

    /**
     * The last time metadata was successfully updated.
     */
    public synchronized long lastSuccessfulUpdate() {
        return this.lastSuccessfulRefreshMs;
    }

    /**
     * The metadata refresh backoff in ms
     */
    public long refreshBackoff() {
        return refreshBackoffMs;
    }

    /**
     * Set state to indicate if metadata for all topics in Kafka cluster is required or not.
     *
     * @param needMetadataForAllTopics boolean indicating need for metadata of all topics in cluster.
     */
    public synchronized void needMetadataForAllTopics(boolean needMetadataForAllTopics) {
        this.needMetadataForAllTopics = needMetadataForAllTopics;
    }

    /**
     * Get whether metadata for all topics is needed or not
     */
    public synchronized boolean needMetadataForAllTopics() {
        return this.needMetadataForAllTopics;
    }

    /**
     * Add a Metadata listener that gets notified of metadata updates
     */
    public synchronized void addListener(Listener listener) {
        this.listeners.add(listener);
    }

    /**
     * Stop notifying the listener of metadata updates
     */
    public synchronized void removeListener(Listener listener) {
        this.listeners.remove(listener);
    }

    /**
     * MetadataUpdate Listener
     */
    public interface Listener {
        void onMetadataUpdate(Cluster cluster);
    }

    private Cluster getClusterForCurrentTopics(Cluster cluster) {
        Set<String> unauthorizedTopics = new HashSet<>();
        Collection<PartitionInfo> partitionInfos = new ArrayList<>();
        List<Node> nodes = Collections.emptyList();
        Set<String> internalTopics = Collections.emptySet();
        String clusterId = null;
        if (cluster != null) {
            clusterId = cluster.clusterResource().clusterId();
            internalTopics = cluster.internalTopics();
            unauthorizedTopics.addAll(cluster.unauthorizedTopics());
            unauthorizedTopics.retainAll(this.topics.keySet());

            for (String topic : this.topics.keySet()) {
                List<PartitionInfo> partitionInfoList = cluster.partitionsForTopic(topic);
                if (partitionInfoList != null) {
                    partitionInfos.addAll(partitionInfoList);
                }
            }
            nodes = cluster.nodes();
        }
        return new Cluster(clusterId, nodes, partitionInfos, unauthorizedTopics, internalTopics);
    }
}
